package job.relevance.anchor_weight;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import mapred.base.TextReducer;

import org.apache.hadoop.io.Text;

public class RLV_AW_AnchorPageRank_Reducer extends TextReducer {

    @Override
    public void reduce(Text key, Iterator<Text> values) throws IOException {
        List<String> list = new ArrayList<String>();
        double pagerank = -1;

        while (values.hasNext()) {
            String v = values.next().toString();
            if (v.startsWith("#")) {
                list.add(v.substring(1));
            } else {
                pagerank = Double.parseDouble(v);
            }
        }

        if (pagerank >= 0) {
            for (String s : list) {
                this.output(new Text(s.replaceFirst("\t", "#").replaceAll("\t",
                        "/***/").trim()), new Text(pagerank + "\t1"));
            }
        }
    }

    @Override
    public void setUp() {
    }

    @Override
    public void tearDown() throws IOException {
    }

}
